package cc.example;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.io.IOException;
import java.time.Duration;

public class FluxDemo {
    public static void main(String[] args) throws IOException {
        Flux.range(1,7)
                .log()
                .subscribe(integer -> {
                    System.out.println();
                });

        // 生成流
        Flux<Integer> just = Flux.just(1, 2, 3, 4, 5);

        // 流不消费没用, 消费就是订阅
        just.subscribe(integer -> {
            System.out.println("e1-" + integer);
        });
// 对于每个消费者流都是一样的,  广播模式
        just.subscribe(integer -> {
            System.out.println("e2-" + integer);
        });

        System.out.println("==================");
        Flux<Long> interval = Flux.interval(Duration.ofSeconds(1));
        interval.subscribe(aLong -> {
            System.out.println(aLong);
        });
        Flux<Object> empty1 = Flux.empty(); // 有一个信号,此时代表完成信号
        // 空流: doOnXXXX  事件感知

        Flux<Integer> empty = Flux.range(1,10)
                .doOnComplete(() -> {
                    System.out.println("流结束了.....");
                })
                .doOnCancel(() -> {
                    System.out.println("流取消了...");
                })
                .doOnNext(o -> {
                    System.out.println("doOnNext " + o);
                });

        empty.subscribe(new BaseSubscriber<Object>() {
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅者发和发布者绑定好");
                request(1);
            }

            @Override
            protected void hookOnNext(Object value) {
                System.out.println("元素到达");
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("数据流完成");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("数据流异常");
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("数据流取消");
            }

            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("数据流 hookFinally");
            }
        });

        System.in.read();


    }
}
